package com.yy.service.impl;


import com.alibaba.fastjson2.JSON;
import com.alibaba.nacos.client.naming.utils.CollectionUtils;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.yy.config.EsImportConfig;
import com.yy.constant.EsConstant;
import com.yy.domain.Prod;
import com.yy.domain.ProdTagReference;
import com.yy.mapper.ProdCommMapper;
import com.yy.mapper.ProdMapper;
import com.yy.mapper.ProdTagReferenceMapper;
import com.yy.model.CommStatistics;
import com.yy.model.ProdEs;
import com.yy.service.ImportService;
import com.yy.utils.AppThreadPoolUtils;
import lombok.extern.log4j.Log4j2;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

@Log4j2
@Service
public class ImportServiceImpl implements ImportService , CommandLineRunner {
    @Resource
    private EsImportConfig esImportConfig;
    @Autowired
    private ProdMapper prodMapper;
    @Autowired
    private RestHighLevelClient restHighLevelClient;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private ProdTagReferenceMapper prodTagReferenceMapper;

    @Autowired
    private ProdCommMapper prodCommMapper;
    /**
     * 上次同步时间
     */
    private Date t;
    /**
     * 日期格式化对象
     */
    private SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss SSS");
    /**
     * 每页的条数，定义好，一般不超过2w
     */
    int size=20;
    /**
     * 全量（分页+多线程）
     * 1.创建索引
     * 2.查总条数
     * 3.计算总页数 （totalPage = 总条数 % size == 0 ? 总条数 / size: ((总条数 / size)+1)）
     * 4.for(int i = 1;i <= totalPage;i++){
     * new Thread(()->{
     * prod: select(1,size);
     * prodEs{prod/tag/comm}
     * bulkRequest(prodEsList)
     * }).start();
     * }
     * --------------- 任务 ------------
     * case   when
     * 深分页如何优化？
     */
    @Override
    public void importAll()  {
        Boolean flag=esImportConfig.getFlag();
        if (flag) {
            //1.创建索引
            createProdEsIndex();
            //2.查总条数     根据时间查询
            Long totalCount = getTotalCount(null);
            if (totalCount == 0) {
                log.info("当前数据库没有合格的数据要导入ES{}", new Date());
                return;
            }
            //3。计算总页数
            int totalPage = (int) Math.ceil(totalCount * 1.0 / size);
            //创建一个减法记数器
            CountDownLatch countDownLatch = new CountDownLatch(totalPage);
            //4.循环导入
            for (int current = 0; current < totalPage; current++) {
                final int pageNumber = current + 1;
                AppThreadPoolUtils.esPoolExecutor.submit(new Runnable() {
                    @Override
                    public void run() {
                        fetchProdToProdEs(pageNumber, size, null);
                        countDownLatch.countDown();//减1
                    }
                });
            }
            //等待countDownLatch到0 ，才能证明上面并行的子线程执行完成，再刷新
            try {
                countDownLatch.await();//阻塞的方法 当countDownLatch归零就是解除了阻塞状态
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //5.刷新索引生成倒排索引
            RefreshRequest refreshRequest = new RefreshRequest(EsConstant.PROD_ES_INDEX);
            try {
                RefreshResponse refresh = this.restHighLevelClient.indices().refresh(refreshRequest, RequestOptions.DEFAULT);
                log.info("手动刷新索引成功:{}" + refresh.getStatus().getStatus());
            } catch (IOException e) {
                e.printStackTrace();
                log.error("手动刷新索引失败:{}" + e.getMessage());
            }
            //6.更改索引的设置副本和索引的自动刷新
            UpdateSettingsRequest updateSettingsRequest = new UpdateSettingsRequest(EsConstant.PROD_ES_INDEX);
            updateSettingsRequest.settings(Settings.builder()
                    .put("number_of_replicas", 2)
                    .put("refresh_interval", "1s")
            );
            AcknowledgedResponse response = null;
            try {
                response = this.restHighLevelClient.indices().putSettings(updateSettingsRequest, RequestOptions.DEFAULT);
            } catch (IOException e) {
                e.printStackTrace();
            }
            if (response.isAcknowledged()) {
                log.info("更改索引设置成功:{}" + new Date());
            }
            //7.记录当前完成的时间
             t = new Date();
            //8.把时间放到Redis里面
            stringRedisTemplate.opsForValue().set(EsConstant.UPDATE_IMPORT_TIME_KEY, sdf.format(t));
            log.info("全量导入数据成功，共导入{}条数据,共{}页,每页{}条", totalCount, totalPage, this.size);
        }
        else {
            log.info("全量导入已关闭{}",new Date());
        }
    }
    /**
     * 增时导入
     */
    @Override
    //Scheduled(initialDelay = 120 * 1000, fixedRate = 120 * 1000)
    public void importUpdate() {

        // 进入就要给时间点
        Date t2 = new Date();
        log.info("增量开始");
        String t1Str = stringRedisTemplate.opsForValue().get(EsConstant.UPDATE_IMPORT_TIME_KEY);
        Date t1 = null;
        try {
            t1 = sdf.parse(t1Str);
        } catch (ParseException e) {
            e.printStackTrace();
        }
        Long totalCount = getTotalCount(t1);
        if (totalCount <= 0L) {
            stringRedisTemplate.opsForValue().set(EsConstant.UPDATE_IMPORT_TIME_KEY, sdf.format(t2));
            log.info("没有商品需要导入");
            return;
        }
        //3.计算总页数
        int totalPage = (int) Math.ceil(totalCount * 1.0 / size);
        //创建一个减法记数器
        CountDownLatch countDownLatch = new CountDownLatch(totalPage);
        //4.循环导入
        for (int current = 0; current < totalPage; current++) {
            final int pageNumber = current + 1;
            AppThreadPoolUtils.esPoolExecutor.submit(new Runnable() {
                @Override
                public void run() {
                    fetchProdToProdEs(pageNumber, size, t);
                    countDownLatch.countDown();//减1
                }
            });
        }
        //等待countDownLatch到0 ，才能证明上面并行的子线程执行完成，再刷新
        try {
            countDownLatch.await();//阻塞的方法 当countDownLatch归零就是解除了阻塞状态
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        stringRedisTemplate.opsForValue().set(EsConstant.UPDATE_IMPORT_TIME_KEY, sdf.format(t2));

    }

    /**
     * 循环处理的方法
     * @param current
     * @param size
     * @param t 时间，增量导入使用
     */
    private void fetchProdToProdEs(int current,int size,Date t){

        //1.分页查询商品数据
        List<Prod> prodList = this.prodMapper.selectByPage(((current - 1) * size), size, t);
        //2.根据商品ID查询商品的活动ID
        List<Long> prodIds = prodList.stream().map(Prod::getProdId).collect(Collectors.toList());
        List<ProdTagReference> prodTagReferencesList = this.prodTagReferenceMapper.selectList(new LambdaQueryWrapper<ProdTagReference>()
                .in(ProdTagReference::getProdId, prodIds)
        );
        //3.把prodTagReferencesList里面的数据按prodId进行组组  prodId-->{12,3,4}
        Map<Long, List<ProdTagReference>> prodTagReferencesListMap = prodTagReferencesList.stream().collect(Collectors.groupingBy(ProdTagReference::getProdId));
        //4.根据商品IDS拿评论数据
        List<CommStatistics> statisticsList = this.prodCommMapper.selectCommStatistics(prodIds);
        //5.把statisticsList转成map<ProdId,CommStatistics>
        Map<Long, CommStatistics> commStatisticsMap = statisticsList.stream().collect(Collectors.toMap(CommStatistics::getProdId, commStatistics -> commStatistics));
        //6.创建一个批处理请求
        BulkRequest bulkRequest = new BulkRequest(EsConstant.PROD_ES_INDEX);
        //7.组合数据算好评率并添加到批处理
        prodList.forEach(prod -> {
            Long prodId = prod.getProdId();
            //7.1创建ESProd对象
            ProdEs prodEs = new ProdEs();
            //7.2把prod里面的数据copy到prodEs
            BeanUtils.copyProperties(prod, prodEs);
            //7.3取出活动数据
            List<ProdTagReference> prodTagReferences = prodTagReferencesListMap.get(prodId);
            if (!CollectionUtils.isEmpty(prodTagReferences)) {
                //取出所有活动ID
                List<Long> tagIds = prodTagReferences.stream().map(ProdTagReference::getTagId).collect(Collectors.toList());
                //放到prodEs
                prodEs.setTagList(tagIds);
            }
            //7.4取出来评论统计数据
            CommStatistics commStatistics = commStatisticsMap.get(prodId);
            if (!ObjectUtils.isEmpty(commStatistics)) {
                Long goodCount = commStatistics.getGoodCount();
                if (!goodCount.equals(0L)) {
                    prodEs.setPraiseNumber(goodCount);
                    //计算好评率
                    BigDecimal positiveRating =
                            new BigDecimal(goodCount).divide(new BigDecimal(commStatistics.getAllCount()), 2, RoundingMode.HALF_UP).multiply(new BigDecimal(100));
                    prodEs.setPositiveRating(positiveRating);
                }
            }
            //7.5向bulkRequest
            IndexRequest indexRequest = new IndexRequest(EsConstant.PROD_ES_INDEX);
            indexRequest.id(prodEs.getProdId().toString()); //用自己的商品ID把ES自动生成的_id给覆盖
            indexRequest.source(JSON.toJSONString(prodEs), XContentType.JSON);
            bulkRequest.add(indexRequest);
        });
        //8.直接批量导入
        BulkResponse bulkResponse = null;
        try {
            bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        log.info("=====第{}页导入完成，这一页共{}条数据 导入结果为:{}", current,prodList.size(), !bulkResponse.hasFailures());
    }




    /**
     * 根据时间查询总条数，如果时间为空，则查询总条数
     * 如果不为空，查询修改之后的数据总条数
     */
    private Long getTotalCount(Date t){
        return prodMapper.selectCount(new LambdaQueryWrapper<Prod>()
                .eq(Prod::getStatus,1)
                .ge(t!=null,Prod::getUpdateTime,t)
        );
    }


    /**
     * 创建索引
     */
    private void createProdEsIndex() {
        CreateIndexRequest request=new CreateIndexRequest(EsConstant.PROD_ES_INDEX);
        //判断这个索引是否存在
        try {
            boolean b = restHighLevelClient.indices().exists(new GetIndexRequest(EsConstant.PROD_ES_INDEX), RequestOptions.DEFAULT);
            if (b) {
                log.error(EsConstant.PROD_ES_INDEX + " 的索引已存在{}", new Date());
                return;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        //配置分片 副本
        request.settings(Settings.builder()
                .put("number_of_shards", 2) //分片数(不能改)
                .put("number_of_replicas", 0)//副本数  为什么给0? 第一次导入时如果有副本，是要同步的。性不能。可以导完之再修改副本
                .put("refresh_interval", "-1") //[关闭倒排索引的刷新]更新索引的频率 如果做数据导入时应该先关闭自己动刷新索引  当导入完成之后再修改刷时间
        );
        //设置映射
        request.mapping("{\n" +
                "    \"properties\": {\n" +
                "        \"positiveRating\": {\n" +
                "            \"type\": \"double\"\n" +
                "        },\n" +
                "        \"tagList\": {\n" +
                "            \"type\": \"long\"\n" +
                "        },\n" +
                "        \"soldNum\": {\n" +
                "            \"type\": \"integer\"\n" +
                "        },\n" +
                "        \"price\": {\n" +
                "            \"type\": \"double\"\n" +
                "        },\n" +
                "        \"prodName\": {\n" +
                "            \"analyzer\": \"ik_max_word\",\n" +
                "            \"type\": \"text\"\n" +
                "        },\n" +
                "        \"praiseNumber\": {\n" +
                "            \"type\": \"long\"\n" +
                "        },\n" +
                "        \"_class\": {\n" +
                "            \"index\": false,\n" +
                "            \"type\": \"keyword\",\n" +
                "            \"doc_values\": false\n" +
                "        },\n" +
                "        \"pic\": {\n" +
                "            \"type\": \"keyword\"\n" +
                "        },\n" +
                "        \"prodId\": {\n" +
                "            \"type\": \"keyword\"\n" +
                "        },\n" +
                "        \"categoryId\": {\n" +
                "            \"type\": \"long\"\n" +
                "        },\n" +
                "        \"shopId\": {\n" +
                "            \"type\": \"long\"\n" +
                "        }\n" +
                "    }\n" +
                "}", XContentType.JSON);

        CreateIndexResponse indexResponse = null;
        try {
            indexResponse = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("创建商品失败:{}", e.getMessage());
        }
        log.info("创建商品索引:{}", indexResponse.isAcknowledged());
    }

    /**
     * 项目启动时执行全量导入
     * @param args
     * @throws Exception
     */
    @Override
    public void run(String... args) throws Exception {
        this.importAll();
    }
}
